package net.wicp.tams.common.others.producer;

import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

import org.apache.kafka.clients.producer.KafkaProducer;

import lombok.extern.slf4j.Slf4j;
import net.wicp.tams.commons.Conf;
import net.wicp.tams.commons.apiext.StringUtil;
import net.wicp.tams.commons.thread.threadlocal.ObjectCreator;

@Slf4j
public class ProducerKafka implements ObjectCreator<KafkaProducer<String, byte[]>> {

	private Map<String, String> nameToType = new HashMap<>();
	{
		nameToType.put("retries", "integer");
		nameToType.put("max.block.ms", "long");
		nameToType.put("max.in.flight.requests.per.connection", "integer");
		nameToType.put("linger.ms", "integer");
		nameToType.put("batch.size", "integer");
		nameToType.put("buffer.memory", "integer");
		nameToType.put("max.request.size", "integer");
		nameToType.put("receive.buffer.bytes", "integer");
		nameToType.put("request.timeout.ms", "integer");
		nameToType.put("send.buffer.bytes", "integer");
		nameToType.put("connections.max.idle.ms", "integer");
	}

	@Override
	public KafkaProducer<String, byte[]> createObject() {
		Map<String, String> paramsMap = Conf.getPre("common.others.kafka.");
		Properties prop = new Properties();
		for (String key : paramsMap.keySet()) {
			if (StringUtil.isNotNull(paramsMap.get(key))) {
				String keytrue = key.replace("common.others.kafka.", "");
				if (nameToType.containsKey(keytrue)) {
					if ("integer".equals(nameToType.get(keytrue))) {
						prop.put(keytrue, Integer.parseInt(paramsMap.get(key)));
					} else if ("long".equals(nameToType.get(keytrue))) {
						prop.put(keytrue, Long.parseLong(paramsMap.get(key).replace("l", "").replace("L", "")));
					}
				} else {
					prop.put(keytrue, paramsMap.get(key));
				}

			}
		}
		log.info("kafka参数:");
		for (Object propele : prop.keySet()) {
			log.info("{}:{}", propele, prop.get(propele));
		}
		// link和batch不用加
		KafkaProducer<String, byte[]> producer = new KafkaProducer<>(prop);
		return producer;
	}
}